package com.xiaoyilin.interceptor;

import com.xiaoyilin.domain.Company;
import com.xiaoyilin.serializer.CompanySerializer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * 不推荐
 * 增加生产者和消费者的耦合度
 * 使用自定义的序列化器
 *
 * @author Administrator
 */
public class Producer {

    public static final String topic = "test";
    public static final String brokerList = "localhost:9092";

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties properties = new Properties();
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, CompanySerializer.class.getName());

        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MyInterceptor.class.getName());
        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ChainInterceptor.class.getName());
        //设置拦截器链
        /* List<String> interceptors = new ArrayList<>();
        interceptors.add(MyInterceptor.class.getName());
        interceptors.add(ChainInterceptor.class.getName());
        //将拦截器加入到配置属性中
        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);
        */
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        KafkaProducer<String, Company> kafkaProducer = new KafkaProducer<>(properties);
        Company company = Company.builder().name("xiaoyilin").address("北京").build();
        ProducerRecord<String, Company> producerRecord = new ProducerRecord<>("test", company);
        kafkaProducer.send(producerRecord).get();
        //关闭生产者释放资源调用该方法后将触发拦截器的close()方法
        kafkaProducer.close();
    }


}
